package com.zmn.oms.business.impl.perf;

import com.alibaba.fastjson.JSONObject;
import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
import com.zmn.common.constant.GlobalConsts;
import com.zmn.oms.business.interfaces.es.OrderWorkEsBService;
import com.zmn.oms.business.interfaces.perf.PerfMasterCategBService;
import com.zmn.oms.common.constant.OrderConsts;
import com.zmn.oms.common.constant.OrderStatusConsts;
import com.zmn.oms.model.entity.perf.PerfMasterCateg;
import com.zmn.oms.services.interfaces.perf.PerfMasterCategService;
import lombok.extern.slf4j.Slf4j;
import org.elasticsearch.index.query.BoolQueryBuilder;
import org.elasticsearch.index.query.QueryBuilders;
import org.elasticsearch.script.Script;
import org.elasticsearch.search.aggregations.AggregationBuilders;
import org.elasticsearch.search.aggregations.Aggregations;
import org.elasticsearch.search.aggregations.bucket.filter.ParsedFilter;
import org.elasticsearch.search.aggregations.bucket.terms.ParsedStringTerms;
import org.elasticsearch.search.aggregations.bucket.terms.Terms;
import org.elasticsearch.search.aggregations.metrics.ParsedCardinality;
import org.elasticsearch.search.aggregations.metrics.ParsedSum;
import org.springframework.data.domain.Pageable;
import org.springframework.data.elasticsearch.core.query.NativeSearchQuery;
import org.springframework.data.elasticsearch.core.query.NativeSearchQueryBuilder;
import org.springframework.stereotype.Service;

import javax.annotation.Resource;
import java.math.BigDecimal;
import java.math.RoundingMode;
import java.time.LocalDate;
import java.time.YearMonth;
import java.time.format.DateTimeFormatter;
import java.time.temporal.ChronoUnit;
import java.util.List;
import java.util.Map;
import java.util.stream.Collectors;
import java.util.stream.Stream;
import static org.elasticsearch.index.query.QueryBuilders.*;

/**
 * 类描述：
 *
 * @author hankangshuai
 * @date 2019/09/30 11:16
 */
@Service
@Slf4j
public class PerfMasterCategBServiceImpl implements PerfMasterCategBService {
    private final String RPC_FAIL_MSG = "远程接口调用失败";
    private final String MASTER_ID_CATEG_ID_TERMS = "MASTER_ID_CATEG_ID_TERMS";
    @Resource
    PerfMasterCategService perfMasterCategService;
    @Resource
    OrderWorkEsBService orderWorkEsBService;

    private Script termsScript = new Script("doc.masterId.value+'|'+doc.showProductGroupId.value");

    /**
     * 按月同步分类绩效
     *
     * @param yearMonth 年月
     * @return 更新数量
     */
    @Override
    public Integer sync(YearMonth yearMonth) {
        LocalDate to = yearMonth.atEndOfMonth();
        LocalDate from = yearMonth.atDay(1);
        // 统计出度量集合
        log.info("perf#master#categ# start statistics from:[{}] to:[{}]", from, to);
        Map<Integer, Map<Integer, Map<String, Integer>>> resultMap = statistics(from, to);
        log.info("perf#master#categ# resultMap [{}]", JSONObject.toJSONString(resultMap));
        // 翻译成实体
        List<PerfMasterCateg> perfMasterCategoryList = resultMapTranslationTOPojo(yearMonth, resultMap);
        log.info("perf#master#categ# perfMasterCategoryList[0] [{}]", perfMasterCategoryList.get(0).toString());

        // 分批插入
        return Lists.partition(perfMasterCategoryList, 300)
                .stream()
                .mapToInt(perfMasterCategService::insertBatch)
                .sum();
    }


    private List<PerfMasterCateg> resultMapTranslationTOPojo(YearMonth yearMonth, Map<Integer, Map<Integer, Map<String, Integer>>> resultMap) {
        return resultMap.entrySet().stream()
                .flatMap(masterIdMapEntry -> {
                    Integer masterId = masterIdMapEntry.getKey();
                    Map<Integer, Map<String, Integer>> categIdMap = masterIdMapEntry.getValue();
                    return categIdMap.entrySet().stream().map(categIdMapEntry -> {
                        Integer categId = categIdMapEntry.getKey();
                        Map<String, Integer> measureMap = categIdMapEntry.getValue();
                        PerfMasterCateg perfMasterCategory = new PerfMasterCateg();
                        perfMasterCategory.setMasterId(masterId);
                        perfMasterCategory.setMonth(Integer.valueOf(yearMonth.format(DateTimeFormatter.ofPattern("yyyyMM"))));
                        perfMasterCategory.setCategId(categId);
                        /*
                         * 客单价 = 订单收入/ 新单量
                         * 成功率 = （成功单量+成功返修单量）/（新单量+成功返修单量)
                         */
                        Integer totalAmountSum = measureMap.get("totalAmountSum");
                        BigDecimal totalAmountSumBig = BigDecimal.valueOf(totalAmountSum);
                        Integer newCount = measureMap.get("newCount");
                        BigDecimal newCountBig = BigDecimal.valueOf(newCount);
                        Integer successCount = measureMap.get("successCount");
                        BigDecimal successCountBig = BigDecimal.valueOf(successCount);
                        Integer successReworkCount = measureMap.get("successReworkCount");
                        BigDecimal successReworkCountBig = BigDecimal.valueOf(successReworkCount);
                        int atv = 0;
                        if (newCount != 0) {
                            atv = totalAmountSumBig.divide(newCountBig, 0, RoundingMode.HALF_UP).intValue();
                        }
                        perfMasterCategory.setAtv(atv);
                        BigDecimal divisor = newCountBig.add(successReworkCountBig);

                        int successRate = 0;
                        if (!divisor.equals(BigDecimal.ZERO)) {
                            successRate = (successCountBig.add(successReworkCountBig)).divide(divisor, 4, RoundingMode.HALF_UP).movePointRight(4).intValue();
                        }
                        perfMasterCategory.setSuccessRate(successRate);
                        return perfMasterCategory;
                    });
                }).collect(Collectors.toList());
    }

    /**
     * @param from 起始日期
     * @param to   结束日期
     * @return 变更条数
     * @see PerfMasterCategBServiceImpl#sync(java.time.YearMonth)
     */
    private Map<Integer, Map<Integer, Map<String, Integer>>> statistics(LocalDate from, LocalDate to) {
        // 基数大于10000不准确，分组统计
        int cardinalNumber = this.getCardinalNumber(from, to);
        log.info("perf#master#categ# cardinal number:[{}] from:[{}] to:[{}]", cardinalNumber, from, to);
        if (cardinalNumber >= 10000) {
            if (from.equals(to)) {
                throw new RuntimeException("统计单日基数超过10000，无法继续统计");
            }
            List<LocalDate> dayList = Stream.iterate(from, date -> date.plusDays(1L))
                    .limit(ChronoUnit.DAYS.between(from, to) + 1)
                    .collect(Collectors.toList());
            //二分后数量
            int halve = BigDecimal.valueOf(dayList.size()).divide(BigDecimal.valueOf(2), 0, BigDecimal.ROUND_CEILING).intValue();

            return Lists.partition(dayList, halve)
                    .stream()
                    .map(localDates -> {
                        // 找到最小值和最大值
                        LocalDate partitionFrom = localDates.get(0);
                        LocalDate partitionTo = localDates.get(localDates.size() - 1);
                        // 递归
                        return statistics(partitionFrom, partitionTo);
                    })
                    // 合并结果集
                    .reduce(Maps.newHashMap(), (resultMap, partitionResultMap) -> {
                        partitionResultMap.forEach((masterId, categIdMap) -> resultMap.merge(masterId, categIdMap, (categIdMap1, categIdMap2) -> {
                            categIdMap2.forEach((categId, measureMap) -> categIdMap1.merge(categId, measureMap, (measureMap1, measureMap2) -> {
                                measureMap2.forEach((measureName, measureValue) -> measureMap1.merge(measureName, measureValue, Integer::sum));
                                return measureMap1;
                            }));
                            return categIdMap1;
                        }));
                        return resultMap;
                    });

        }
        NativeSearchQuery buildQuery = buildQuery(from, to);
        Aggregations aggregations = orderWorkEsBService.getAggregationsByQuery(buildQuery);
        return this.resultsExtractor(aggregations);
    }

    private NativeSearchQuery buildQuery(LocalDate from, LocalDate to) {
        return new NativeSearchQueryBuilder().withPageable(Pageable.unpaged())
                .withQuery(getFilter(from, to))
                .addAggregation(
                        AggregationBuilders.terms(MASTER_ID_CATEG_ID_TERMS)
                                .script(termsScript)
                                .size(10000)
                                /*
                                 * 客单价 = 订单收入/ 新单量
                                 * 成功率 = （成功单量+成功返修单量）/（新单量+成功返修单量)
                                 */
                                .subAggregation(
                                        // 订单收入
                                        AggregationBuilders.filter("completeTime", rangeQuery("completeTime").from(from.toString()).to(to.toString()))
                                                .subAggregation(AggregationBuilders.sum("totalAmountSum").field("totalAmount"))
                                )
                                .subAggregation(
                                        AggregationBuilders.filter("distributeTime", rangeQuery("distributeTime").from(from.toString()).to(to.toString()))
                                                .subAggregation(
                                                        //新单量
                                                        AggregationBuilders.filter(
                                                                "newCount",
                                                                boolQuery()
                                                                        .must(termQuery("type", OrderConsts.ORDER_TYPE_NEW))
                                                                        .mustNot(boolQuery()
                                                                                .must(termQuery("resultStatus", OrderStatusConsts.WORK_RESULT_DISCARD))
                                                                                .must(termQuery("cancelerType", GlobalConsts.OPERATE_USER_TYPE_SYSTEM)))
                                                        )
                                                )
                                                .subAggregation(
                                                        //成功单量
                                                        AggregationBuilders.filter(
                                                                "successNewCount",
                                                                boolQuery()
                                                                        .must(termQuery("type", OrderConsts.ORDER_TYPE_NEW))
                                                                        .must(termQuery("resultStatus", OrderStatusConsts.WORK_RESULT_SUCCESS)))
                                                )
                                                .subAggregation(
                                                        // 成功返修单量
                                                        AggregationBuilders.filter(
                                                                "successReworkCount",
                                                                boolQuery()
                                                                        .must(termQuery("type", OrderConsts.ORDER_TYPE_REWORK))
                                                                        .must(termQuery("resultStatus", OrderStatusConsts.WORK_RESULT_SUCCESS))
                                                        )
                                                )
                                )

                )
                .build();
    }

    private int getCardinalNumber(LocalDate from, LocalDate to) {
        NativeSearchQuery build = new NativeSearchQueryBuilder()
                .withPageable(Pageable.unpaged())
                .withQuery(getFilter(from, to))
                .addAggregation(
                        AggregationBuilders.cardinality(MASTER_ID_CATEG_ID_TERMS).script(termsScript)
                ).build();
        Aggregations aggregations = orderWorkEsBService.getAggregationsByQuery(build);
        ParsedCardinality internalCardinality = aggregations.get(MASTER_ID_CATEG_ID_TERMS);
        Long cardinalNumber = internalCardinality.getValue();
        if (cardinalNumber > Integer.MAX_VALUE) {
            throw new RuntimeException("统计基数过大!无法继续统计");
        }
        return cardinalNumber.intValue();
    }

    private Map<Integer, Map<Integer, Map<String, Integer>>> resultsExtractor(Aggregations aggregations) {
        ParsedStringTerms masterIdCategIdTerms = aggregations.get(MASTER_ID_CATEG_ID_TERMS);
        List<? extends Terms.Bucket> masterIdCategIdBuckets = masterIdCategIdTerms.getBuckets();

        return masterIdCategIdBuckets.stream()
                .collect(Collectors.groupingBy(bucket -> {
                            String masterIdCategIdKey = bucket.getKeyAsString();
                            String[] masterIdCategIdArr = masterIdCategIdKey.split("\\|", 2);
                            return Integer.parseInt(masterIdCategIdArr[0]);
                        },
                        Collectors.toMap(bucket -> {
                                    String masterIdCategIdKey = bucket.getKeyAsString();
                                    String[] masterIdCategIdArr = masterIdCategIdKey.split("\\|", 2);
                                    return Integer.parseInt(masterIdCategIdArr[1]);
                                },
                                bucket -> {
                                    Aggregations bucketAggregations = bucket.getAggregations();
                                    // 订单收入
                                    int totalAmountSum = (int) ((ParsedSum) ((ParsedFilter) bucketAggregations.get("completeTime")).getAggregations().get("totalAmountSum")).getValue();
                                    // 新单量
                                    int newCount = (int) ((ParsedFilter) ((ParsedFilter) bucketAggregations.get("distributeTime")).getAggregations().get("newCount")).getDocCount();
                                    // 成功单量
                                    int successCount = (int) ((ParsedFilter) ((ParsedFilter) bucketAggregations.get("distributeTime")).getAggregations().get("successNewCount")).getDocCount();
                                    // 成功返修单量
                                    int successReworkCount = (int) ((ParsedFilter) ((ParsedFilter) bucketAggregations.get("distributeTime")).getAggregations().get("successReworkCount")).getDocCount();
                                    Map<String, Integer> measureMap = Maps.newHashMapWithExpectedSize(4);
                                    measureMap.put("totalAmountSum", totalAmountSum);
                                    measureMap.put("newCount", newCount);
                                    measureMap.put("successCount", successCount);
                                    measureMap.put("successReworkCount", successReworkCount);
                                    return measureMap;
                                })));
    }

    private BoolQueryBuilder getFilter(LocalDate from, LocalDate to) {
        return boolQuery()
                .filter(termQuery("plat", GlobalConsts.PLAT_MARK_ZMN))
                .filter(termQuery("platWork", GlobalConsts.PLAT_MARK_ZMN))
                .filter(termQuery("isDuplicate", false))
                .filter(boolQuery()
                        .should(QueryBuilders.rangeQuery("distributeTime").from(from.toString()).to(to.toString()))
                        .should(QueryBuilders.rangeQuery("completeTime").from(from.toString()).to(to.toString()))
                );
    }
}
